package com.yzg.pool;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author misterWei
 * @create 2020年06月07号:11点53分
 * @mailbox forcontinue@163.com
 *   自定义线程池
 */
@Slf4j(topic = "c.ThreadPoolTest")
public class ThreadPoolTest {

    public static void main(String[] args) {
        ThreadPool threadPool = new ThreadPool(2,1000,TimeUnit.MICROSECONDS,1
                ,(queue,task)->{
            // 1. 死等
            //queue.add(task);
            // 2) 带超时等待
           // queue.addWait(task, 1500, TimeUnit.MILLISECONDS);
            // 3) 让调用者放弃任务执行
 //log.debug("放弃{}", task);
            // 4) 让调用者抛出异常
 throw new RuntimeException("任务执行失败 " + task);
            // 5) 让调用者自己执行任务
           // task.run();


        });
        for (int i = 0; i < 5; i++) {
            int j = i;
            threadPool.execute(()->{
                log.info("正在执行...{}",j);
            });
        }
    }

    /**
     * 功能: 线程池 配合自定义队列实现
     */
    static class ThreadPool{
        // 存放任务的队列
        private BlockingQueue<Runnable> runnables;

        //线程池的总核心数
        private int core;

        //队列的长度
        private int queueSize;

        //线程集合
        private Set<Worker> workers = new HashSet<>();

        //超时时间
        private long time;

        //超时单位
        private TimeUnit timeUnit;

        //自定义策率
        private RejectStrategy rejectStrategy;

        public ThreadPool(int core, long time, TimeUnit timeUnit,int queueSize,RejectStrategy<Runnable> rejectStrategy) {
            this.core = core;
            this.time = time;
            this.timeUnit = timeUnit;
            this.queueSize = queueSize;
            this.runnables = new BlockingQueue<Runnable>(this.queueSize);
            this.rejectStrategy = rejectStrategy;
        }

        public void execute(Runnable task){
            synchronized (workers){
                if (workers.size() < core){
                    //进行执行
                    Worker worker = new Worker(task);
                    workers.add(worker);
                    log.info("正在执行{}",worker);
                    worker.start();
                }else{
                    //加入任务队列等待
                    runnables.customizeAdd(rejectStrategy,task);
                }
            }



        }

        class Worker extends Thread{
            private Runnable runnable;

            public Worker(Runnable runnable) {
                this.runnable = runnable;
            }
            @Override
            public void run() {
                while (runnable != null || (runnable = runnables.pollTime(time,timeUnit)) != null){
                   log.info("当前任务正在执行..{}",runnable);
                    runnable.run();
                    runnable = null;
                }
                synchronized (workers){
                    log.debug("worker 被移除{}", this);
                    workers.remove(this);
                }

            }
        }
    }


    /**
     * 阻塞&非阻塞 任务队列
     * 功能介绍: 用来存放任务的队列
     */
    @Slf4j(topic = "c.BlockingQueue")
    static
    class BlockingQueue<T> {

        //队列
        private Deque<T> queues = new ArrayDeque<T>();

        //非公平锁
        private ReentrantLock lock = new ReentrantLock();

        //条件变量 针对 拿 进行阻塞 (消费者)
        Condition consumer =  lock.newCondition();

        //条件变量 针对 拿 进行阻塞 (生产者)
        Condition producer =  lock.newCondition();

        //任务总数
        private int taskSize;

        public BlockingQueue(int taskSize) {
            this.taskSize = taskSize;
        }

        /**
         * 超时策率的获取任务
         *
         */
        public T pollTime(long time, TimeUnit timeUnit){
            lock.lock();
            try {
                long l  = timeUnit.toNanos(time);
                while (queues.isEmpty()){
                    try {
                        log.info("任务队列为空,正在有时间等待...{}",queues.size());
                        if (l <=0 )
                            return null;
                        l = consumer.awaitNanos(l);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                T task = queues.removeFirst();
                producer.signal();
                return task;
            }finally {
                lock.unlock();
            }
        }


        /**
         * 获取任务的方式
         */
        public T poll(){
            lock.lock();
            try {
                while (queues.isEmpty()){
                    try {
                        log.info("任务队列为空,正在等待...{}",queues.size());
                        consumer.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                T task = queues.removeFirst();
                producer.signal();
                return task;
            }finally {
                lock.unlock();
            }
        }

        /**
         * 添加任务的方式
         */
        public void add(T task){
            lock.lock();
            try {
                while (queues.size() >= taskSize){
                    try {
                        log.info("当前任务以达到阀值 size:{},taskSize{}",queues.size(),taskSize);
                        producer.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                queues.addLast(task);
                consumer.signal();
            }finally {
                lock.unlock();
            }
        }


        /**
         * 添加任务有时间的等待方式
         */
        public void addWait(T task,long time,TimeUnit timeUnit){
            lock.lock();
            try {
                long nanos = timeUnit.toNanos(time);
                while (queues.size() >= taskSize){
                    try {
                        log.info("当前任务以达到阀值 size:{},taskSize{}",queues.size(),taskSize);
                        if (nanos <= 0) {
                            return;
                        }
                        nanos =  producer.awaitNanos(nanos);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                queues.addLast(task);
                consumer.signal();
            }finally {
                lock.unlock();
            }
        }

        /**
         * 自定义的拒绝策率的执行
         */
        public void customizeAdd(RejectStrategy<T> strategy,T task){
            lock.lock();
            try {
                if (queues.size() == taskSize) {
                    //自定义策略实现
                    strategy.reject(this,task);
                }else{
                    queues.addLast(task);
                    consumer.signal();
                }

            }finally {
                lock.unlock();
            }

        }

    }
}

/**
 * 策率自定义实现
 * @param <T>
 */
@FunctionalInterface
interface RejectStrategy<T>{
  void reject(ThreadPoolTest.BlockingQueue<T> queue,T task);
}
